package com.gy.hadoop.mr.secondarysort;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.net.URI;

/**
 * 按照工资倒叙  工资一样按照年龄升序
 * hadoop jar  kkb-bigdata-1.0-SNAPSHOT.jar com.gy.hadoop.mr.secondarysort.SecondarySort /data/input/sort/word /data/output/mr/sort
 */
public class SecondarySort {

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration conf = new Configuration();
        conf.set("mapreduce.app-submission.cross-platform", "true");
        Job job = Job.getInstance(conf, SecondarySort.class.getSimpleName());
        job.setJar("target/kkb-bigdata-1.0-SNAPSHOT.jar");

        FileSystem fs = FileSystem.get(URI.create(args[1]), conf);
        if (fs.exists(new Path(args[1])))
            fs.delete(new Path(args[1]), true);

        FileInputFormat.setInputPaths(job, new Path(args[0]));
        job.setMapperClass(MyMap.class);
        job.setMapOutputKeyClass(Person.class);
        job.setMapOutputValueClass(NullWritable.class);

        job.setNumReduceTasks(1);

        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        job.setReducerClass(MyReduce.class);
        job.setOutputKeyClass(Person.class);
        job.setOutputValueClass(NullWritable.class);


        job.waitForCompletion(true);


    }

    /**
     * tom	20	8000
     * nancy	22	8000
     * ketty	22	9000
     * stone	19	10000
     * green	19	11000
     * white	30	29000
     * socrates	29	40000
     */
    public static class MyMap extends Mapper<LongWritable, Text, Person, NullWritable> {

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String[] values = value.toString().split("\t");
            String name = values[0];
            Integer age = Integer.parseInt(values[1]);
            Integer salary = Integer.parseInt(values[2]);

            Person person = new Person(name, age, salary);
            context.write(person, NullWritable.get());
        }

    }

    public static class MyReduce extends Reducer<Person, NullWritable, Person, NullWritable> {

        @Override
        protected void reduce(Person key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
            context.write(key, NullWritable.get());
        }

    }


}
